package com.simon.cdc.serializable;


import com.fasterxml.jackson.databind.ObjectMapper;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;


public class CustomJsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    /**
     * 从SourceRecord对象中反序列化数据
     *
     * @param sourceRecord 数据源记录，用于提取数据
     * @param out          输出收集器，用于收集和输出处理后的数据
     * @throws Exception 如果反序列化过程中发生错误，则抛出异常
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> out) throws Exception {
        // 确保SourceRecord不为空
        Objects.requireNonNull(sourceRecord, "SourceRecord cannot be null");
        // 获取数据值，并强制转换为Struct类型
        Struct value = (Struct) sourceRecord.value();
        // 如果值为null，则直接返回
        if (value == null) {
            return;
        }

        // 初始化结果映射，用于存储反序列化的数据
        Map<String, Object> result = new HashMap<>();
        // 解析操作类型
        Envelope.Operation op = Envelope.operationFor(sourceRecord);
        // 将操作类型代码（如c, u, d, r）添加到结果映射中
        result.put("op", op.code().toLowerCase()); // c, u, d, r
        // 解析主键
        parsePrimaryKeys(sourceRecord, result);

        // 解析数据变更内容
        parseDataChanges(value, result);

        // 解析源信息
        parseSourceInfo(value, result);

        // 序列化为JSON并输出
        out.collect(OBJECT_MAPPER.writeValueAsString(result));
    }

    /**
     * 获取当前类产生的类型信息
     * <p>
     * 此方法重写了接口或父类的方法，用于指定当前类或上下文环境中产生的数据类型
     * 在数据处理和类型定义中，这个方法用于明确指定返回类型为字符串（String）的类型信息
     *
     * @return TypeInformation<String> 返回字符串类型的类型信息对象，用于描述和处理数据流或数据结构的类型
     */
    @Override
    public TypeInformation<String> getProducedType() {
        return Types.STRING;
    }

    /**
     * 解析主键信息
     * 该方法用于解析给定记录中的主键信息，并将其添加到结果映射中
     * 主键信息对于唯一标识记录非常重要，并且可以在数据处理和存储时用于各种目的
     *
     * @param record 一个包含键信息的源记录，预期是数据库记录或其他数据源的快照
     * @param result 一个映射，用于存储解析后的主键信息，方法执行结果将添加到这个映射中
     */
    private void parsePrimaryKeys(SourceRecord record, Map<String, Object> result) {
        // 检查记录的键是否为Struct类型，这是进一步处理的前提
        if (record.key() instanceof Struct keyStruct) {
            // 获取键的架构，用于遍历其字段
            Schema keySchema = record.keySchema();
            // 创建一个映射来存储主键信息
            Map<String, Object> pk = new HashMap<>();
            // 遍历键架构中的所有字段
            for (Field field : keySchema.fields()) {
                // 将每个字段及其对应值添加到主键映射中
                pk.put(field.name(), keyStruct.get(field));
            }
            // 将主键信息添加到结果映射中，以便进一步处理或存储
            result.put("pk", pk);
        }
    }

    /**
     * 解析数据变更内容
     * 该方法主要用于解析数据变更记录中的before和after状态
     * 它通过将结构化数据转换为Map对象，更新到结果集中
     *
     * @param value  包含数据变更信息的结构体
     * @param result 存储解析后数据变更内容的Map对象
     */
    private void parseDataChanges(Struct value, Map<String, Object> result) {
        // 获取变更后数据结构
        Struct afterStruct = value.getStruct("after");
        // 如果变更后数据存在，则将其转换为Map并添加到结果集中
        if (afterStruct != null) {
            result.put("after", structToMap(afterStruct));
        }

        // 获取变更前数据结构
        Struct beforeStruct = value.getStruct("before");
        // 如果变更前数据存在，则将其转换为Map并添加到结果集中
        if (beforeStruct != null) {
            result.put("before", structToMap(beforeStruct));
        }
    }

    /**
     * 解析源信息
     * 此方法旨在从给定的结构体中提取并解析源信息，然后将其添加到结果映射中
     *
     * @param value  包含源信息的结构体
     * @param result 用于存储解析后的源信息的映射
     */
    private void parseSourceInfo(Struct value, Map<String, Object> result) {
        // 获取名为"source"的结构体
        Struct sourceStruct = value.getStruct("source");
        // 如果获取到的结构体不为空，则将其转换为映射并添加到结果中
        if (sourceStruct != null) {
            result.put("source", structToMap(sourceStruct));
        }
    }

    /**
     * 将Struct对象转换为Map对象
     * 此方法用于将Struct对象中的字段和值转换成Map的形式，便于进一步处理或查询
     *
     * @param struct 需要转换的Struct对象，包含了一组字段和相应的值
     * @return 返回一个Map对象，其中包含了Struct对象中的所有字段和值
     */
    private Map<String, Object> structToMap(Struct struct) {
        // 创建一个HashMap对象，用于存储Struct对象中的字段和值
        Map<String, Object> map = new HashMap<>();
        // 遍历Struct对象中的所有字段
        for (Field field : struct.schema().fields()) {
            // 将字段名和对应的值放入Map中
            map.put(field.name(), struct.get(field));
        }
        // 返回转换后的Map对象
        return map;
    }
}

